package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.self.annotation.Cacheable;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.search.factory.ScheduleTaskThreadFactory;
import com.atguigu.tingshu.search.repository.AlbumInfoRepository;
import com.atguigu.tingshu.search.repository.SuggestInfoRepository;
import com.atguigu.tingshu.search.runable.RebuildDistroBloomFilterRunnable;
import com.atguigu.tingshu.search.service.ItemService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import com.google.common.hash.BloomFilter;
import jakarta.annotation.PostConstruct;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class ItemServiceImpl implements ItemService {

    @Autowired
    private AlbumInfoRepository albumInfoRepository;

    @Autowired
    private SuggestInfoRepository suggestInfoRepository;

    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private UserInfoFeignClient userInfoFeignClient;

    static ExecutorService executorService = null;

    static {
        executorService = Executors.newFixedThreadPool(4);
    }

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private ItemServiceImpl itemServiceImpl;

    @Autowired
    private RBloomFilter rBloomFilter;
    HashMap<String, Object> localCache = new HashMap<>();

    // 深度获取锁标记
    ThreadLocal<String> deepAcquisitionLockMap = new ThreadLocal<>();


    // 布隆过滤器
    private static BloomFilter<Long> longBloomFilter;

    /**
     * 构建布隆过滤器
     */
    @PostConstruct
    public void rebuildBloomFilter() {
        //线上 （从7天后凌晨2点钟开始执行第一次） 以后每隔7天在执行下一次
        ScheduleTaskThreadFactory instance = ScheduleTaskThreadFactory.getInstance();
        // 获取当前时间戳，并计算距离7天后凌晨2点钟的毫秒数差值
        long taskBeginTime = instance.diffTime(System.currentTimeMillis());
        // 创建一个定时任务，在指定时间后执行指定的任务
        instance.execute(
                // 创建一个线程，执行指定的任务
                new RebuildDistroBloomFilterRunnable(redissonClient, redisTemplate,itemServiceImpl), taskBeginTime, TimeUnit.MILLISECONDS);

    }

    /**
     * 获取所有专辑id
     *
     * @return
     */
    public List<Long> getAlbumIds() {
        // 远程调用获取所有专辑id
        Result<ArrayList<Long>> albumIdsResult = albumInfoFeignClient.getAllAlbumInfoIds();
        List<Long> albumIdsData = albumIdsResult.getData();
        // 判断集合是否为空
        if (!CollectionUtils.isEmpty(albumIdsData)) {
            // 不为空返回专辑id集合
            return albumIdsData;
        }
        // 如果集合为空，则返回一个空的列表
        return new ArrayList<>();
    }
    /**
     * 分布式缓存+分布式布隆过滤器
     *
     * @param albumId
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private Map opsDistroCacheAndBloomFilter(Long albumId) throws InterruptedException, ExecutionException {

        //获取当前时间
        Long startTime = System.currentTimeMillis();

        // 0.查询分布式布隆过滤器
        boolean contains = rBloomFilter.contains(albumId);
        // 如果布隆过滤器中不包含该id，则返回null
        if (!contains) {
            return null;
        }

        // 1.查询分布式缓存
        String cacheKey = "cache:info:albumId:" + albumId;
        String jsonFromCache = redisTemplate.opsForValue().get(cacheKey);

        // 2.缓存命中
        if (!StringUtils.isEmpty(jsonFromCache)) {
            // 返回缓存命中的数据
            System.out.println("分布式缓存命中耗时ms：" + (System.currentTimeMillis() - startTime));
            //返回序列化后的数据
            return JSONObject.parseObject(jsonFromCache, Map.class);
        }
        // 3.缓存未命中 回源（从网络找 从文件中 分布式组件找（minio） 数据库）
        Map<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);

        // 4.同步数据到缓存
        redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoFromDb));

        //获取结束时间
        Long endTime = System.currentTimeMillis();
        System.out.println("分布式缓存未命中耗时ms：" + (endTime - startTime));

        // 返回数据
        return albumInfoFromDb;

    }

    /**
     * 本地缓存：第一次查询：301ms  第二次查询0ms  第三次查询0ms  0ms  平均在0ms左右
     *
     * @param albumId
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */

    private Map opsLocalCache(Long albumId) throws InterruptedException, ExecutionException {

        Long startTime = System.currentTimeMillis();
        String cacheKey = "cache:info:albumId:" + albumId;
        // 1.查询本地缓存
        boolean b = localCache.containsKey(cacheKey);

        // 2.缓存命中
        if (b) {
            System.out.println("本地缓存命中耗时ms：" + (System.currentTimeMillis() - startTime));
            return (Map) localCache.get(cacheKey);
        }

        // 3.缓存未命中 回源
        Map<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);

        // 4.将数据同步到缓存中
        localCache.put(cacheKey, albumInfoFromDb);
        System.out.println("本地缓存未命中耗时ms：" + (System.currentTimeMillis() - startTime));
        return albumInfoFromDb;
    }

    @Override
    @SneakyThrows
    public void offAlbumEs(Long albumId) {
        albumInfoRepository.deleteById(albumId);
    }



    @Override
    @SneakyThrows
    @Cacheable(cacheKey = RedisConstant.CACHE_INFO_PREFIX + "#{#args[0]}",
            distroLockKey = RedisConstant.ALBUM_LOCK_SUFFIX + "#{#args[0]}",
            distroBloomKey = "#{#args[0]}",
            distroLockSwitch = true,
            distroBloomSwitch = true
    )
    public Map<String, Object> getAlbumInfoDetail(Long albumId) {
//        return opsDistroCacheAndCacheHitV1(albumId);
        return getAlbumInfoFromDb(albumId);
    }

    /**
     * 分布式缓存+解决缓存击穿使用分布式锁（Redis版本） 不带自动续期功能
     * <p>
     * 问题一：抢到锁的线程在执行业务期间非常耗时，导致抢到锁的线程这个客户端不能执行delete释放锁，导致redis中锁的key一直存在（死锁）
     * 解决方案：双端释放锁：客户端执行delete释放锁 服务端通过expire设置过期时间。保证死锁现象最大可能被避免掉。
     * <p>
     * 问题二：setkey和expirekey不是原子操作，极端情况下 依然会出现死锁。
     * 解决方案：将设置锁key的过期时间和设置锁key作为一个原子操作setIfAbsent四个参数API SetIfAbsent四个参数的底层是将设置key和给key设置过期时间在一个连接中执行（原子操作）
     * <p>
     * 问题三：一旦服务端给锁设置过期时间，在高并发下有可能出现锁的误删：场景：A B一起执行 A线程可能会删除B线程加的锁
     * 解决方案：
     * 1、每个线程加锁的时候都有一个属于自己得我锁标识
     * 2、每个线程删除锁的时候判断锁是不是自己加的。
     * <p>
     * 问题四：判断锁和删除锁不是原子操作 在极端情况下 判断锁相等，但是删除锁的时候别的线程加上 导致误删现象又出现。
     * 解决方案：将判断锁和删除锁通过Lua脚本保证原子操作。解决极端下锁的误删
     * <p>
     * <p>
     * 栈空间不够；出现的stackoverflow(栈溢出)
     * 堆空间不够：oom(out of  memory)
     * <p>
     * <p>
     * <p>
     * <p>
     * redis版本的分布式自旋可重入锁+递归+threadlocal+双缓存机制的缓存击穿解决方案
     *
     * @param albumId
     * @return
     */
    @SneakyThrows
    private HashMap opsDistroCacheAndCacheHitV1(Long albumId) {
        HashMap<String, Object> albumInfoMap;
        // 构建缓存键，用于从Redis中获取或存储该专辑的信息
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        // 构建锁键，确保同一时间只有一个线程可以更新指定专辑的信息
        String lockKey = RedisConstant.CACHE_LOCK_SUFFIX + albumId;
        // 标记是否成功获取到分布式锁，默认为false
        Boolean lockFlag = false;
        // 存储当前操作的锁值
        String lockValue = null;
        // 尝试从Redis获取与锁键相关的值，检查是否有其他线程正在处理相同专辑的信息
        String cacheValue = redisTemplate.opsForValue().get(cacheKey);
        if (!StringUtils.isEmpty(cacheValue)) { // 如果锁键对应的值存在且不为空
            // 直接返回解析后的缓存数据（假设之前已经有其他线程存储了这个专辑的信息）
            return JSONObject.parseObject(cacheValue, HashMap.class);
        }

        // 从当前线程的ThreadLocal变量中获取深度获取锁的值（每个线程独立）
        String deepAcquisitionLockValue = deepAcquisitionLockMap.get();
        if (!StringUtils.isEmpty(deepAcquisitionLockValue)) { // 如果深度获取锁的值存在
            lockFlag = true; // 表示已经获得了锁
            lockValue = deepAcquisitionLockValue; // 设置当前锁值为此深度获取锁的值
        } else {
            // 生成一个新的UUID作为锁值，并去除其中的"-"符号
            lockValue = UUID.randomUUID().toString().replaceAll("-", "");
            // 尝试以原子方式设置锁键和锁值，有效期为60秒。如果设置成功，则表示获得了锁
            lockFlag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 60, TimeUnit.SECONDS);
        }

        if (Boolean.TRUE.equals(lockFlag)) { // 如果成功获得了锁
            try {
                // 从数据库中获取专辑信息
                albumInfoMap = getAlbumInfoFromDb(albumId);
                // 将获取到的专辑信息序列化为JSON字符串并存储到Redis缓存中
                redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoMap));

                // Lua脚本：仅当锁值匹配时才删除锁，确保安全释放锁
                String script = """
                        if redis.call('GET',KEYS[1])==ARGV[1] then
                            return redis.call('del',KEYS[1])
                        else
                            return 0
                        end
                        """;
                // 执行Lua脚本尝试释放锁
                Long execute = redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), List.of(lockKey), lockValue);
                if (execute == 0) {
                    log.info("释放锁成功");
                } else {
                    log.error("释放锁失败");
                }
            } finally {
                // 为防止内存泄露
                // 移除当前线程的ThreadLocal变量中的锁值
                deepAcquisitionLockMap.remove();
            }
            return albumInfoMap; // 返回获取到的专辑信息
        } else { // 如果未能成功获得锁
            // 策略一：查询缓存  策略二:递归  混合使用：策略一（查询缓存）+策略二（自旋+递归）
            // 只有等锁释放了 我去递归才有意义 否则别递归 自旋抢锁
            // 通过自旋抢锁 解决无效递归 可以经可能减少栈溢出风险   cpu会很快被打满。
            Thread.sleep(200);  // 压测给一个最终时间
            String result = redisTemplate.opsForValue().get(cacheKey);
            // 解决百分之99的正常情况
            // 一次查询缓存：为了解决正常情况下抢不到锁的线程从缓存中直接查询数据返回
            if (!StringUtils.isEmpty(result)) {
                return JSONObject.parseObject(result, HashMap.class);
            } else {
                while (true) {
                    // 尝试从Redis获取缓存的专辑信息
                    String redisValue = redisTemplate.opsForValue().get(cacheKey);
                    if (!StringUtils.isEmpty(redisValue)) { // 如果找到了缓存数据
                        // 解析并返回缓存中的专辑信息
                        return JSONObject.parseObject(redisValue, HashMap.class);
                    }
                    // 再次尝试获取锁
                    Boolean isLock = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 60, TimeUnit.SECONDS);
                    if (Boolean.TRUE.equals(isLock)) {
                        // 设置当前线程的ThreadLocal变量中的锁值
                        deepAcquisitionLockMap.set(lockValue);
                        break; // 退出循环
                    }
                }
                //缓存未命中。
                //当前线程未能成功获取 Redis 锁。
                //自旋获取锁失败后，递归调用该方法以继续尝试获取锁。
                return opsDistroCacheAndCacheHitV1(albumId);
            }
        }
    }

    /**
     * 分布式缓存+解决缓存击穿使用分布式锁（Redis版本） 自带自动续期功能
     * 守护线程特点：只要jvm中还存在用户线程，守护线程会一直工作。只要jvm中没有用户线程 守护线程就不在工作
     *
     * @param albumId
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @SneakyThrows
    private HashMap opsDistroCacheAndCacheHitV2(Long albumId) {
        //1.构建缓存键和锁键
        // 1.1构建缓存键，用于从Redis中获取或存储该专辑的信息
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        // 1.2构建锁键，确保同一时间只有一个线程可以更新指定专辑的信息
        String lockKey = RedisConstant.CACHE_LOCK_SUFFIX + albumId;
        // 1.3标记是否成功获取到分布式锁，默认为false
        Boolean lockFlag = false;
        String lockValue = "";

        String cacheValue = redisTemplate.opsForValue().get(cacheKey);
        //3.判断缓存是否命中
        if (!StringUtils.isEmpty(cacheValue)) {
            // 如果锁键对应的值存在且不为空
            // 直接返回解析后的缓存数据（假设之前已经有其他线程存储了这个专辑的信息）
            return JSONObject.parseObject(cacheValue, HashMap.class);
        }

        // 从当前线程的ThreadLocal变量中获取深度获取锁的值（每个线程独立）
        String deepAcquisitionLockValue = deepAcquisitionLockMap.get();
        if (!StringUtils.isEmpty(deepAcquisitionLockValue)) { // 如果深度获取锁的值存在
            lockFlag = true; // 表示已经获得了锁
            lockValue = deepAcquisitionLockValue; // 设置当前锁值为此深度获取锁的值
        } else {
            // 生成一个新的UUID作为锁值，并去除其中的"-"符号
            lockValue = UUID.randomUUID().toString().replaceAll("-", "");
            // 尝试以原子方式设置锁键和锁值，有效期为60秒。如果设置成功，则表示获得了锁
            lockFlag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 60, TimeUnit.SECONDS);
        }

        if (Boolean.TRUE.equals(lockFlag)) {
            // 启动一个守护线程进行锁续期
            Thread renewalThread = getRenewalThread(lockKey, lockValue, 30L, 60L);
            HashMap<String, Object> albumInfoMap;
            try {
                // 从数据库中获取专辑信息
                albumInfoMap = getAlbumInfoFromDb(albumId);
                redisTemplate.opsForValue().set(cacheKey, JSONObject.toJSONString(albumInfoMap));
                // Lua脚本：仅当锁值匹配时才删除锁，确保安全释放锁
                String script = """
                       if redis.call('GET',KEYS[1])==ARGV[1] then
                           return redis.call('del',KEYS[1])
                       else
                           return 0
                       end
                       """;
                Long execute = redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), List.of(lockKey), lockValue);
                if (execute != null && execute == 1) {
                    log.info("释放锁成功");
                } else {
                    log.error("释放锁失败");
                }
            } finally {
                //中断守护线程
                renewalThread.interrupt();
            }
            return albumInfoMap; // 返回获取到的专辑信息
        } else { // 如果未能成功获得锁
            // 策略一：查询缓存  策略二:递归  混合使用：策略一（查询缓存）+策略二（自旋+递归）
            // 只有等锁释放了 我去递归才有意义 否则别递归 自旋抢锁
            // 通过自旋抢锁 解决无效递归 可以经可能减少栈溢出风险   cpu会很快被打满。
            Thread.sleep(200);  // 压测给一个最终时间
            String result = redisTemplate.opsForValue().get(cacheKey);
            // 解决百分之99的正常情况
            // 一次查询缓存：为了解决正常情况下抢不到锁的线程从缓存中直接查询数据返回
            if (!StringUtils.isEmpty(result)) {
                return JSONObject.parseObject(result, HashMap.class);
            } else {
                while (true) {
                    // 尝试从Redis获取缓存的专辑信息
                    String redisValue = redisTemplate.opsForValue().get(cacheKey);
                    if (!StringUtils.isEmpty(redisValue)) { // 如果找到了缓存数据
                        // 解析并返回缓存中的专辑信息
                        return JSONObject.parseObject(redisValue, HashMap.class);
                    }
                    // 再次尝试获取锁
                    Boolean isLock = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 60, TimeUnit.SECONDS);
                    if (Boolean.TRUE.equals(isLock)) {
                        // 设置当前线程的ThreadLocal变量中的锁值
                        deepAcquisitionLockMap.set(lockValue);
                        break; // 退出循环
                    }
                }
                //缓存未命中。
                //当前线程未能成功获取 Redis 锁。
                //自旋获取锁失败后，递归调用该方法以继续尝试获取锁。
                return opsDistroCacheAndCacheHitV2(albumId);
            }
        }
    }


    /**
     * 获取用于续期锁的线程
     * 该线程会不断检查锁是否需要续期，并在需要时进行续期操作
     *
     * @param lockKey 锁的键，用于标识锁
     * @param lockValue 锁的值，用于验证锁的所有权
     * @return 返回一个负责锁续期任务的线程
     */
    private Thread getRenewalThread(String lockKey, String lockValue, Long sleepTime, Long ttl) {
        // 创建并返回一个线程，该线程负责锁的续期操作

        Thread renewalThread = new Thread(() -> {
            while (true) {
                try {
                    // 每隔一定时间检查并续期锁
                    Thread.sleep(sleepTime); // 假设每30秒检查一次
                    // 检查锁是否仍然由当前线程持有，以决定是否需要续期
                    if (redisTemplate.opsForValue().get(lockKey).equals(lockValue)) {
                        // 续期锁，设置锁的过期时间为60秒
                        redisTemplate.expire(lockKey, ttl, TimeUnit.SECONDS);
                    } else {
                        // 锁已被其他线程占用，退出续期
                        break;
                    }
                } catch (InterruptedException e) {
                    // 捕获中断异常，并重新设置当前线程的中断状态
                    Thread.currentThread().interrupt();
                    // 退出续期线程
                    break;
                }
            }
        }, "Lock-Renewal-Thread");
        // 通过调用setDaemon(true)将线程标记为守护线程。
        // 这意味着当JVM中所有的非守护线程都结束后，守护线程会自动结束，不会阻止应用程序的退出。
        renewalThread.setDaemon(true);
        renewalThread.start();
        // 返回初始化好的续期线程
        return renewalThread;
    }



    /**
     * 异步数据获取：通过使用CompletableFuture类来异步执行多个任务，每个任务负责从不同的微服务中获取特定类型的专辑相关信息。
     * 线程池指定：所有异步任务都指定了同一个线程池threadPoolExecutor来执行，以控制并发任务的执行。
     * 结果整合：在所有异步任务完成后，将获取到的数据整合到一个HashMap对象中并返回。
     *
     * @param albumId 专辑ID，用于查询专辑相关信息
     * @return 包含专辑详细信息的Map对象
     * @throws InterruptedException 如果线程被中断
     * @throws ExecutionException 如果CompletableFuture执行失败
     */
    @NotNull
    private HashMap<String, Object> getAlbumInfoFromDb(Long albumId) throws InterruptedException, ExecutionException {

        long startTime = System.currentTimeMillis();
        // 1.创建一个HashMap对象，用于存储从各个微服务获取到的数据
        HashMap<String, Object> result = new HashMap<>();

        // 2.使用CompletableFuture.supplyAsync异步执行任务，获取专辑的基本信息
        CompletableFuture<Long> longCompletableFuture = CompletableFuture.supplyAsync(() -> {
            // 打印当前执行线程的信息
            System.out.println("执行查询专辑的基本信息使用的线程：" + Thread.currentThread().getName());

            // 远程调用专辑微服务，获取专辑的基本信息
            Result<AlbumInfo> alumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);

            // 获取返回结果中的数据部分
            AlbumInfo alumInfoData = alumInfoResult.getData();

            // 检查返回的数据是否为空，如果为空则抛出异常
            Assert.notNull(alumInfoData, "远程查询专辑微服务获取专辑的基本信息失败");

            // 将专辑的基本信息存入result中
            result.put("albumInfo", alumInfoData);  // 专辑的基本信息

            // 返回专辑创建者的用户ID
            return alumInfoData.getUserId();
        }, threadPoolExecutor); // 指定线程池执行此异步任务

        // 3.使用CompletableFuture.runAsync异步执行任务，获取专辑的统计信息
        CompletableFuture<Void> albumInfoResultFuture = CompletableFuture.runAsync(() -> {
            // 打印当前执行线程的信息
            System.out.println("执行查询专辑的统计信息使用的线程：" + Thread.currentThread().getName());

            // 远程调用专辑微服务，获取专辑的统计信息
            Result<AlbumStatVo> albumInfoResult = albumInfoFeignClient.getAlbumInfoStat(albumId);

            // 获取返回结果中的数据部分
            AlbumStatVo albumInfoStatData = albumInfoResult.getData();

            // 检查返回的数据是否为空，如果为空则抛出异常
            Assert.notNull(albumInfoStatData, "远程查询专辑微服务获取专辑的统计信息失败");

            // 将专辑的统计信息存入result中
            result.put("albumStatVo", albumInfoStatData); // 专辑的统计信息
        }, threadPoolExecutor); // 指定线程池执行此异步任务

        // 4.使用CompletableFuture.runAsync异步执行任务，获取专辑的分类信息
        CompletableFuture<Void> albumInfoCategoryResultFuture = CompletableFuture.runAsync(() -> {
            // 打印当前执行线程的信息
            System.out.println("执行查询专辑的分类信息使用的线程：" + Thread.currentThread().getName());

            // 远程调用专辑微服务，获取专辑的分类信息
            Result<BaseCategoryView> albumInfoCategoryResult = albumInfoFeignClient.getAlbumInfoCategory(albumId);

            // 获取返回结果中的数据部分
            BaseCategoryView albumInfoCategoryData = albumInfoCategoryResult.getData();

            // 检查返回的数据是否为空，如果为空则抛出异常
            Assert.notNull(albumInfoCategoryData, "远程查询专辑微服务获取专辑的分类信息失败");

            // 将专辑的分类信息存入result中
            result.put("baseCategoryView", albumInfoCategoryData); // 专辑所在的分类信息
        }, threadPoolExecutor); // 指定线程池执行此异步任务

        // 5.当获取到用户ID后，使用CompletableFuture.thenAcceptAsync异步执行任务，获取用户的详细信息
        CompletableFuture<Void> userInfoResultFuture = longCompletableFuture.thenAcceptAsync((userId) -> {
            // 打印当前执行线程的信息
            System.out.println("执行查询用户信息使用的线程：" + Thread.currentThread().getName());

            // 远程调用用户微服务，获取用户的详细信息
            Result<UserInfoVo> userInfoResult = userInfoFeignClient.getUserInfo(userId);

            // 获取返回结果中的数据部分
            UserInfoVo userInfoData = userInfoResult.getData();

            // 检查返回的数据是否为空，如果为空则抛出异常，并附带错误原因
            Assert.notNull(userInfoData, "远程查询用户微服务获取用户信息失败,原因:" + userInfoResult.getMessage());

            // 将用户的详细信息存入result中
            result.put("announcer", userInfoData); // 专辑对应的主播信息
        }, threadPoolExecutor); // 指定线程池执行此异步任务

        // 6.使用CompletableFuture.allOf等待所有异步任务完成
        CompletableFuture<Void> future = CompletableFuture.allOf(
                longCompletableFuture,
                albumInfoResultFuture,
                albumInfoCategoryResultFuture,
                userInfoResultFuture
        ); // 等待所有线程把任务都做完
        try {
            // 调用future.get()阻塞当前工作线程，直到所有异步任务线程完成各自任务
            future.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        System.out.println("纯异步耗时ms：" + (System.currentTimeMillis() - startTime));

        // 返回包含专辑详细信息的Map对象
        return result;
    }



    @Override
    @SneakyThrows
    public void preHitRank() {

        // 1.获取所有的一级分类id
        Result<List<Long>> c1IdListResult = albumInfoFeignClient.getCategory1IdList();
        List<Long> c1IdListData = c1IdListResult.getData();
        if (CollectionUtils.isEmpty(c1IdListData)) {
            throw new GuiguException(201, "平台不存在一级分类id");
        }
        // 2.遍历一级分类id集合
        for (Long c1Id : c1IdListData) {
            String[] dimensions = new String[]{"hotScore", "playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum"};
            for (String dimension : dimensions) {
                SearchRequest searchRequest = SearchRequest.of(b -> b.index("albuminfo").query(qb -> qb.term(tmqb -> tmqb.field("category1Id").value(c1Id))).sort(sob -> sob.field(fsb -> fsb.field(dimension).order(SortOrder.Desc))).size(10));
                SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);

                ArrayList<AlbumInfoIndex> albumInfoIndices = new ArrayList<>();
                for (Hit<AlbumInfoIndex> hit : response.hits().hits()) {
                    AlbumInfoIndex source = hit.source();
                    albumInfoIndices.add(source);
                }
                String bigKey = RedisConstant.RANKING_KEY_PREFIX + c1Id;
                // 3.存储到Redis中(String  List  Set Zset  Hash(非常合适))
                redisTemplate.opsForHash().put(bigKey, dimension, JSONObject.toJSONString(albumInfoIndices));
            }

        }
    }

    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long c1Id, String dimension) {
        String bigKey = RedisConstant.RANKING_KEY_PREFIX + c1Id;
        String resultStr = (String) redisTemplate.opsForHash().get(bigKey, dimension);
        if (StringUtils.isEmpty(resultStr)) {
            return new ArrayList<AlbumInfoIndexVo>();
        }
        List<AlbumInfoIndexVo> albumInfoIndexVoList = JSONObject.parseArray(resultStr, AlbumInfoIndexVo.class);
        return albumInfoIndexVoList;
    }


    /**
     *
     * 优化方式2 使用线程池+countdownlatch使用
     * 耗时：第一次：380ms 第二次：53ms  平均在50ms
     * 为什么第一次比较耗时：后面不怎么耗时：
     * 1、线程池来说，第一批任务进来需要创建线程（创建线程会耗时 但是耗时不多）
     * 2、远程调用的时候缓存机制。
     * 当客户端第一次远程调用服务端时，openfeign的底层会从远程的注册中心上获取对方的信息。（缓存到客户端本地）而当客户端后面每次调用服务端时，直接从客户端本地内存（Map）中获取  不在请求远端注册中心。
     * 服务发现（OpenFeign采用了缓存思想）缓存是最能提高性能。异步提高效率很低。多线程上下文切换（很耗时） Redis(单线程模型解决读写操作)、
     * Redis高性能的原因：
     * 1、Redis采用了单线程模型
     * 2、Redis客户端采用多路复用机制 select/epoll模型
     * 3、Redis内部拥有高效的数据结构（Hash 树  跳跃表 ）
     * 4、Redis中的数据都保存在内存中。
     *
     * @param albumId
     */
    @Override
    public void onAlbumEs(Long albumId) {
        CountDownLatch countDownLatch = new CountDownLatch(4);  // 局部变量(不能作为一个成员变量 不然在多线程情况下出现数据不一致问题)

        // 1.创建文档对象
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
        Long startTime = System.currentTimeMillis();
        // 2.给文档对象赋值
        Future<Long> longFuture = executorService.submit(() -> {
            // 2.1 远程查询专辑微服务获取专辑的基本信息
            Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);
            AlbumInfo albumData = albumInfoResult.getData();
            if (albumData == null) {
                throw new GuiguException(201, "搜索微服务调用专辑微服务获取专辑基本信息失败");
            }
            albumInfoIndex.setId(albumData.getId());   // 文档id(专辑id)
            albumInfoIndex.setAlbumTitle(albumData.getAlbumTitle()); // 专辑标题
            albumInfoIndex.setAlbumIntro(albumData.getAlbumIntro()); // 专辑的简介
            albumInfoIndex.setCoverUrl(albumData.getCoverUrl()); // 专辑的封面图片地址
            albumInfoIndex.setIncludeTrackCount(albumData.getIncludeTrackCount()); // 专辑包含的声音集数
            albumInfoIndex.setIsFinished(albumData.getIsFinished().toString()); // 专辑是否完结
            albumInfoIndex.setPayType(albumData.getPayType()); // 专辑付费类型（三种）
            List<AttributeValueIndex> attributeValueIndexList = albumData.getAlbumAttributeValueVoList().stream().map(albumAttributeValue -> {
                AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId());
                attributeValueIndex.setValueId(albumAttributeValue.getValueId());
                return attributeValueIndex;

            }).collect(Collectors.toList());
            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList); // 专辑的标签
            albumInfoIndex.setCreateTime(new Date()); // 专辑在es中的创建时间
            countDownLatch.countDown(); //计数器的值-1
            return albumData.getUserId();
        });

        executorService.execute(() -> {
            // 2.2 远程查询用户微服务获取用户的基本信息
            try {
                Long userId = longFuture.get(); // 获取上一个任务的结果 会阻塞当前线程
                Result<UserInfoVo> userInfoVoResult = userInfoFeignClient.getUserInfo(userId);
                UserInfoVo userData = userInfoVoResult.getData();
                if (userData == null) {
                    throw new GuiguException(201, "搜索微服务调用用户微服务获取用户的基本信息失败");
                }
                albumInfoIndex.setAnnouncerName(userData.getNickname()); // 专辑对应主播的昵称
                countDownLatch.countDown(); //计数器的值-1
            } catch (Exception e) {
                log.error("当前线程阻塞期间出现了异常：{}", e.getMessage());
            }

        });

        executorService.execute(() -> {
            // 2.3 远程查询专辑微服务获取专辑的分类信息
            Result<BaseCategoryView> baseCategoryViewResult = albumInfoFeignClient.getAlbumInfoCategory(albumId);
            BaseCategoryView baseCategoryViewData = baseCategoryViewResult.getData();
            if (baseCategoryViewData == null) {
                throw new GuiguException(201, "搜索微服务调用专辑微服务获取专辑分类信息失败");
            }
            albumInfoIndex.setCategory1Id(baseCategoryViewData.getCategory1Id()); // 专辑所在的一级分类（id）
            albumInfoIndex.setCategory2Id(baseCategoryViewData.getCategory2Id());// 专辑所在的二级分类（id）
            albumInfoIndex.setCategory3Id(baseCategoryViewData.getCategory3Id());// 专辑所在的三级分类（id）
            countDownLatch.countDown(); //计数器的值-1

        });
        executorService.execute(() -> {
            // 2.4 远程查询专辑微服务获取专辑的统计信息
            Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumInfoStat(albumId);
            AlbumStatVo albumStatVoData = albumStatVoResult.getData();
            if (albumStatVoData == null) {
                throw new GuiguException(201, "搜索微服务调用专辑微服务获取专辑统计信息失败");
            }
            Integer playStatNum = albumStatVoData.getPlayStatNum();
            albumInfoIndex.setPlayStatNum(playStatNum);  // 专辑的播放量
            Integer subscribeStatNum = albumStatVoData.getSubscribeStatNum();
            albumInfoIndex.setSubscribeStatNum(subscribeStatNum); // 专辑的订阅量
            Integer buyStatNum = albumStatVoData.getBuyStatNum();
            albumInfoIndex.setBuyStatNum(buyStatNum); // 专辑的购买量
            Integer commentStatNum = albumStatVoData.getCommentStatNum();
            albumInfoIndex.setCommentStatNum(commentStatNum); // 专辑的评论数

//        Double hotScore = playStatNum * 0.4 + subscribeStatNum * 0.1 + buyStatNum * 0.3 + commentStatNum * 0.2;
            Double hotScore = new Random().nextDouble();
            albumInfoIndex.setHotScore(hotScore);
            countDownLatch.countDown(); //计数器的值-1
        });

        // 3.将文档对象存储到Es中
        try {
            countDownLatch.await(); // 让当前线程也即tomcat线程池的线程阻塞
            AlbumInfoIndex flag = albumInfoRepository.save(albumInfoIndex);   // tomcat线程池的线程执行
            Long endTime = System.currentTimeMillis();
            log.info("专辑：{}存储到Es中：{}:耗时：{}ms", albumId, (flag != null)? "success" : "fail", endTime - startTime);
        } catch (InterruptedException e) {
            log.error("当前阻塞期间出现了异常:{}", e.getMessage());
        }

        //4.创建一个提示词的文档 放到提示词索引库下
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(albumId.toString());
        suggestIndex.setTitle(albumInfoIndex.getAlbumTitle());
        suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()}));
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())}));
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())}));
        suggestInfoRepository.save(suggestIndex);

    }

}
